/********************************************************/
/*	Copyright (C) 2012 Gong Li Bin			*/
/*	Project:	MapReduce			*/
/*	Author:		gong_libin			*/
/*	Date:		2016_06_01			*/
/*	File:		MrUtils.cpp			*/
/********************************************************/

#include "MrUtils.h"

extern CMrSet g_CPoll;

void MrJbQuit(int iSignal)
{
	MR_ERROR("Program Stopped by signal[%d]\n", iSignal);

	exit(MR_FAILURE);
}

void MrJbResponse(char* pszUrl)
{
	CMrUdp CUdp;
	CMrPro CPro;
	UCHAR uszHead[MR_PRO_HEAD] = { 0 };

	CPro.MrProResponse(uszHead);
	if (MR_SUCCESS == CUdp.MrNetOpen(MR_SOCK_DGRAM)) {
		CUdp.MrUdpSetAddr(pszUrl);
		CUdp.MrNetSend(uszHead, MR_PRO_HEAD);
		CUdp.MrNetClose();
	}
	return;
}

void MrJbResponse(struct sockaddr_in* pstAddr)
{
	CMrUdp CUdp;
	CMrPro CPro;
	UCHAR uszHead[MR_PRO_HEAD] = { 0 };

	CPro.MrProResponse(uszHead);
	if (MR_SUCCESS == CUdp.MrNetOpen(MR_SOCK_DGRAM)) {
		CUdp.MrUdpSetAddr(pstAddr);
		CUdp.MrNetSend(uszHead, MR_PRO_HEAD);
		CUdp.MrNetClose();
	}

	return;
}

void MrJbHelp(char* pszProc, char* pszVer)
{
	MR_PRINT("Usage: %s [options]\n", pszProc);
	MR_PRINT("Valid options are:\n");
	MR_PRINT("[*] -c or --conf:       set configure.\n");
	MR_PRINT("[0] -t or --thread:     set thread total.\n");
	MR_PRINT("[0] -d or --debug:      set debug mode.\n");
	MR_PRINT("[0] -h or --help:       show help.\n");
	MR_PRINT("[0] -v or --version:    show version.\n");
	MR_PRINT("Copyright (C) 2012 Gong Li Bin Technology Inc.\n");
	MR_PRINT("%s Build on %s %s.\n", pszVer, __DATE__, __TIME__);

	return;
}

void* MrJbAdminCore(void* pNULL)
{
	CMrPro CPro;
	CMrUdp CUdp;
	UINT uiCount = 0;
	UCHAR uszSend[MR_PRO_HEAD] = { 0 };
	UCHAR uszRecv[MR_PRO_HEAD] = { 0 };

	if (MR_SUCCESS == CUdp.MrNetOpen(MR_SOCK_DGRAM)) {
		CPro.MrProRequest(uszSend);
		CUdp.MrNetSetTimeOut(0, MR_USEC);
		while (true) {
			uiCount = 0;
			while (uiCount < g_CPoll.MrSetGetCount()) {
				CUdp.MrUdpSetAddr((void*)&((MRJBPOLL_S*)g_CPoll.MrSetIndexGet(uiCount))->m_stHost.m_unAddr.m_stAddr);
				if (CUdp.MrNetSend(uszSend, MR_PRO_HEAD) > 0) {
					if (CUdp.MrNetRecv(uszRecv, MR_PRO_HEAD) > 0) {
						CPro.MrProReset();
						CPro.MrProParse(uszRecv, MR_PRO_HEAD);
						if (MR_DEF_LENG == CPro.MrProGetResult()) {
							((MRJBPOLL_S*)g_CPoll.MrSetIndexGet(uiCount))->m_bStatus = true;
						}
					}
					else {
						((MRJBPOLL_S*)g_CPoll.MrSetIndexGet(uiCount))->m_bStatus = false;
					}
				}
				memset(uszRecv, '\0', MR_PRO_HEAD);
				uiCount += 1;
			}
			sleep(MR_UT_SLEEP);
		}
		CUdp.MrNetClose();
	}

	pthread_exit(NULL);
}

int MrJbPortInit(CMrDb& rCDb, char* pszSql, MRCONFIG_S* pstConf, char* pszPort)
{
	MRPROKV_S* pstKv = NULL;
	int iReturn = MR_SUCCESS;
	char szSql[MR_PACKET] = { 0 };

	if (NULL != (pstKv = MrProKvMalloc(MR_BUFFER, MR_PACKET))) {
		sprintf(szSql, pszSql, pstConf->m_szDbase, pstConf->m_szTable, pstConf->m_szID);
		if (rCDb.MrDbSend(MR_DEF_SELT, szSql) > 0) {
			if (MR_SUCCESS == rCDb.MrDbRecv(pstKv)) {
				memcpy(pszPort, (pstKv + 0)->m_puszValue, (pstKv + 0)->m_iLength);
			}
			else {
				MR_ERROR("Failed to MrDbRecv\n");
				iReturn = MR_FAILURE;
			}
		}
		else {
			MR_ERROR("Failed to MrDbSend\n");
			iReturn = MR_FAILURE;
		}

		MrProKvFree(&pstKv, MR_BUFFER);
	}
	else {
		MR_ERROR("Failed to MrProKvMalloc\n");
		iReturn = MR_FAILURE;
	}

	return iReturn;
}

void MrJbDevDisplay(UINT* puiDev)
{
	USHORT usDevice = 0;

	MR_PRINT("Dev:");
	while (0 != *(puiDev + usDevice)) {
		MR_PRINT(" %d", *(puiDev + usDevice));
		usDevice += 1;
	}
	MR_PRINT("\n");

	return;
}

void MrJbDevInput(UINT* puiDev, CMrSet& CDev)
{
	USHORT usDev = 0;

	while (0 != (*(puiDev + usDev) = ((MRJBHOST_S*)CDev.MrSetIndexGet(usDev))->m_uiID)) usDev += 1;

	return;
}

int MrJbDevInit(CMrDb& rCDb, char* pszSql, MRCONFIG_S* pstConf, CMrSet* pCSet, UCHAR ucType)
{
	int iSize = 0;
	UINT uiCount = 0;
	void* pDev = NULL;
	MRJBPOLL_S stPoll;
	MRPROKV_S* pstKv = NULL;
	int iReturn = MR_SUCCESS;
	char szSql[MR_PACKET] = { 0 };
	char szValue[MR_VALUE] = { 0 };

	switch (ucType) {
		case MR_JB_POLL:
			pDev = (void*)&stPoll;
			iSize = sizeof(MRJBPOLL_S);
			pCSet->MrSetCompare(MrJbCompPoll);
			break;
		case MR_JB_HOST:
			iSize = sizeof(MRJBHOST_S);
			pDev = (void*)&stPoll.m_stHost;
			pCSet->MrSetCompare(MrJbCompHost);
			break;
		default:
			break;
	}

	if (NULL != (pstKv = MrProKvMalloc(MR_BUFFER, MR_PACKET))) {
		sprintf(szSql, pszSql, pstConf->m_szDbase, pstConf->m_szTable);
		if (rCDb.MrDbSend(MR_DEF_SELT, szSql) > 0) {
			if (MR_SUCCESS == rCDb.MrDbRecv(pstKv)) {
				if (MR_SUCCESS == pCSet->MrSetInit(MR_JB_DEVICE, iSize)) {
					stPoll.m_bStatus = true;
					stPoll.m_stHost.m_unAddr.m_stAddr.sin_family = AF_INET;
					while ('\0' != (pstKv + uiCount)->m_ucKey) {
						/* ID */
						memcpy(szValue, (pstKv + uiCount)->m_puszValue, (pstKv + uiCount)->m_iLength);
						szValue[(pstKv + uiCount)->m_iLength] = '\0';
						stPoll.m_stHost.m_uiID = atoi(szValue);
						uiCount += 1;
						/* Addr */
						memcpy(szValue, (pstKv + uiCount)->m_puszValue, (pstKv + uiCount)->m_iLength);
						szValue[(pstKv + uiCount)->m_iLength] = '\0';
						stPoll.m_stHost.m_unAddr.m_stAddr.sin_addr.s_addr = inet_addr(szValue);
						uiCount += 1;
						/* port */
						memcpy(szValue, (pstKv + uiCount)->m_puszValue, (pstKv + uiCount)->m_iLength);
						szValue[(pstKv + uiCount)->m_iLength] = '\0';
						stPoll.m_stHost.m_unAddr.m_stAddr.sin_port = htons(atoi(szValue));
						uiCount += 1;
						/* insert */
						pCSet->MrSetIndexPut(pCSet->MrSetGetCount(), pDev);
					}
					pCSet->MrSetSort();
				}
				else {
					MR_ERROR("Failed to MrSetInit\n");
					iReturn = MR_FAILURE;
				}
			}
			else {
				MR_ERROR("Failed to MrDbRecv\n");
				iReturn = MR_FAILURE;
			}
		}
		else {
			MR_ERROR("Failed to MrDbSend\n");
			iReturn = MR_FAILURE;
		}

		MrProKvFree(&pstKv, MR_BUFFER);
	}
	else {
		MR_ERROR("Failed to MrProKvMalloc\n");
		iReturn = MR_FAILURE;
	}

	return iReturn;
}

void MrJbDevReqAry(MRJBPACKET_S* pstPacket, UINT* puiDev)
{
	CMrPro CPro;
	USHORT usCount = 0;
	int iLength = MR_PRO_HEAD;
	char szValue[MR_VALUE] = { 0 };
	MRPROKV_S stKv = {MR_JB_ID, 0, (UCHAR*)szValue};
	MRPROHEAD_S stHead = {
		MR_PRO_FLAG,
		MR_PRO_1,
		MR_PRO_CTRL,
		MR_PRO_0,
		MR_PRO_Y,
		MR_PRO_0,
		MR_DEF_POLL,
		MR_DEF_SELT,
		MR_JB_AVG,
		MR_DEF_SUCC,
		MR_PRO_0,
		MR_PRO_0,
	};

	while (0 != *(puiDev + usCount)) {
		sprintf(szValue, "%d", *(puiDev + usCount));
		stKv.m_iLength = strlen(szValue);

		iLength += MrProKvOutput(&stKv, MR_PRO_1, pstPacket->m_uszPacket + iLength, MR_BIGPACKET - iLength);
		memset(szValue, '\0', MR_VALUE);
		usCount += 1;
	}

	CPro.MrProPutBody(NULL);
	CPro.MrProInduce(&stHead);
	CPro.MrProPutTotal(usCount);
	CPro.MrProPutLength(iLength - MR_PRO_HEAD);
	CPro.MrProCreate(pstPacket->m_uszPacket, MR_BIGPACKET);
	pstPacket->m_iLength = MR_PRO_HEAD + CPro.MrProGetLength();

	return;
}

void MrJbDevReqSet(MRJBPACKET_S* pstPacket, CMrSet& rCDev)
{
	CMrPro CPro;
	UINT uiCount = 0;
	int iLength = MR_PRO_HEAD;
	char szValue[MR_VALUE] = { 0 };
	MRPROKV_S stKv = {MR_JB_ID, 0, (UCHAR*)szValue};
	MRPROHEAD_S stHead = {
		MR_PRO_FLAG,
		MR_PRO_1,
		MR_PRO_CTRL,
		MR_PRO_0,
		MR_PRO_Y,
		MR_PRO_0,
		MR_DEF_POLL,
		MR_DEF_SELT,
		MR_JB_AVG,
		MR_DEF_SUCC,
		MR_PRO_0,
		MR_PRO_0,
	};

	while (uiCount < rCDev.MrSetGetCount()) {
		sprintf(szValue, "%d", ((MRJBHOST_S*)rCDev.MrSetIndexGet(uiCount))->m_uiID);
		stKv.m_iLength = strlen(szValue);

		iLength += MrProKvOutput(&stKv, MR_PRO_1, pstPacket->m_uszPacket + iLength, MR_BIGPACKET - iLength);
		memset(szValue, '\0', MR_VALUE);
		uiCount += 1;
	}

	CPro.MrProPutBody(NULL);
	CPro.MrProInduce(&stHead);
	CPro.MrProPutTotal(rCDev.MrSetGetCount());
	CPro.MrProPutLength(iLength - MR_PRO_HEAD);
	CPro.MrProCreate(pstPacket->m_uszPacket, MR_BIGPACKET);
	pstPacket->m_iLength = MR_PRO_HEAD + CPro.MrProGetLength();

	return;
}

UCHAR MrJbDevSort(CMrUdp& rCSend, MRJBPACKET_S* pstPacket, UINT* puiHost)
{
	CMrPro CPro;
	int iLength = 0;
	UINT uiBegin = 0;
	UINT uiCursor = 0;
	USHORT usHost = 0;
	MRPROKV_S stKv[MR_JB_PAIR];
	UCHAR ucReturn = MR_DEF_UNKN;
	char szValue[MR_VALUE] = { 0 };
	UCHAR uszPacket[MR_BIGPACKET] = { 0 };

	uiBegin = uiCursor = rand() % g_CPoll.MrSetGetCount();

	do {
		if (true == ((MRJBPOLL_S*)g_CPoll.MrSetIndexGet(uiCursor))->m_bStatus) {
			rCSend.MrUdpSetAddr((void*)&((MRJBPOLL_S*)g_CPoll.MrSetIndexGet(uiCursor))->m_stHost.m_unAddr.m_stAddr);
			if (rCSend.MrNetSend(pstPacket->m_uszPacket, pstPacket->m_iLength) > 0) {
				if ((iLength = rCSend.MrNetRecv(uszPacket, MR_BIGPACKET)) > 0) {
					CPro.MrProParse(uszPacket, iLength);
					if (MR_DEF_SUCC == (ucReturn = CPro.MrProGetResult())) {
						memset(stKv, '\0', sizeof(MRPROKV_S) * MR_JB_PAIR);
						MrProKvInput(stKv, CPro.MrProGetTotal(), CPro.MrProGetBody(), CPro.MrProGetLength());
						while (usHost < CPro.MrProGetTotal()) {
							memcpy(szValue, stKv[usHost].m_puszValue, stKv[usHost].m_iLength);
							szValue[stKv[usHost].m_iLength] = '\0';
							*(puiHost + usHost) = atoi(szValue);
							usHost += 1;
						}
					}
					else if (MR_DEF_NULL == ucReturn) {
						break;
					}
					else {
					}
				}
				else {
					ucReturn = MR_DEF_RECV;
				}
			}
			else {
				ucReturn = MR_DEF_SEND;
			}
		}
		if (++ uiCursor == g_CPoll.MrSetGetCount()) {
			uiCursor = 0;
		}
		if (uiCursor == uiBegin) {
			break;
		}
	} while (MR_DEF_SUCC != ucReturn);

	return ucReturn;
}

int MrJbCompPoll(const void* pDev1, const void* pDev2)
{
	return ((MRJBPOLL_S*)pDev1)->m_stHost.m_uiID - ((MRJBPOLL_S*)pDev2)->m_stHost.m_uiID;
}

int MrJbCompHost(const void* pDev1, const void* pDev2)
{
	return ((MRJBHOST_S*)pDev1)->m_uiID - ((MRJBHOST_S*)pDev2)->m_uiID;
}
